package cn.ds.hbase.utils;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 项目名称:hbase-demo<br>
 * 包名:cn.ds.hbase.utils<br>
 * 用于:hbase-demo<br>
 * 创建时间:2019年01月24日<br>
 * 更新时间:2019年01月24日<br>
 *
 * @author :lds（创建人）<br>
 * @version :v1.0（版本号）<br>
 * @since jdk1.8
 */
@Slf4j
public class HBaseUtil {

    public static boolean createTable(String tName,String... cols) throws IOException {
        final TableName tableName = TableName.valueOf(tName);
        try(Admin admin = HBsaeConn.getHBaseConn().getAdmin()) {
            if (admin.tableExists(tableName)){
                log.error("表【{}】已经存在！",tName);
                return false;
            }
            if (cols==null||cols.length<1){
                log.error("创建的表没有列数据");
                return false;
            }
            final HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);
            for (String col : cols) {
                final HColumnDescriptor columnDescriptor = new HColumnDescriptor(col);
                tableDescriptor.addFamily(columnDescriptor);
            }
            admin.createTable(tableDescriptor);
        }

        return true;
    }

    public static boolean modifyTable(String tName,String... cols){

        return false;
    }

    public static boolean deleteTable(String tName) throws IOException {
       try( Admin admin = HBsaeConn.getHBaseConn().getAdmin()) {
           final TableName tableName = TableName.valueOf(tName);
           if (admin.tableExists(tableName)){
               admin.disableTable(tableName);
               admin.deleteTable(tableName);
               return true;
           }else {
               log.error("表【{}】不存在",tName);
               return false;
           }
       }
    }

    public static List<String> getTableNames() throws IOException {
        try( Admin admin = HBsaeConn.getHBaseConn().getAdmin()) {
            final HTableDescriptor[] hTableDescriptors = admin.listTables();
            if (hTableDescriptors==null||hTableDescriptors.length<1){
                return Collections.emptyList();
            }

           return Arrays.stream(hTableDescriptors).map(HTableDescriptor::getNameAsString).collect(Collectors.toList());
        }
    }

    public static boolean insertRow(String tName,String rowKey,String colFamily,String col,String val) throws IOException {
       try(Table table = HBsaeConn.getTable(tName)) {
           final Put put = new Put(Bytes.toBytes(rowKey));
           put.addColumn(Bytes.toBytes(colFamily),Bytes.toBytes(col),Bytes.toBytes(val));
           table.put(put);
       }
        return true;
    }

    public static void deleteRow(String tName,String rowKey,String colFamily,String col) throws IOException {
        try(Table table = HBsaeConn.getTable(tName)) {
            final Delete delete = new Delete(Bytes.toBytes(rowKey));
            if (StringUtils.isNotBlank(colFamily)){
                delete.addFamily(Bytes.toBytes(colFamily));
                if (StringUtils.isNotBlank(col)){
                    delete.addColumn(Bytes.toBytes(colFamily),Bytes.toBytes(col));
                }
            }
            table.delete(delete);
        }
    }

    public static List<JSONObject> getData(String tName,String rowKey,String colFamily,String col) throws IOException {
        try(Table table = HBsaeConn.getTable(tName)){
            final Get get = new Get(Bytes.toBytes(rowKey));
            get.addColumn(Bytes.toBytes(colFamily),Bytes.toBytes(col));
            final Result result = table.get(get);
            final Cell[] cells = result.rawCells();

            final List<JSONObject> objectList = Arrays.stream(cells).map(cell -> {
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("rowName", new String(CellUtil.cloneRow(cell)));
                jsonObject.put("family",new String(CellUtil.cloneFamily(cell)));
                jsonObject.put("row",new String(CellUtil.cloneQualifier(cell)));
                jsonObject.put("value",new String(CellUtil.cloneValue(cell)));
                return jsonObject;
            }).collect(Collectors.toList());

            return objectList;
        }
    }


}
